package me.illtamer.rpc.transport.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import me.illtamer.rpc.entity.RpcRequest;
import me.illtamer.rpc.entity.RpcResponse;
import me.illtamer.rpc.enumeration.RpcError;
import me.illtamer.rpc.exception.RpcException;
import me.illtamer.rpc.factory.SingletonFactory;
import me.illtamer.rpc.loadbalancer.LoadBalancer;
import me.illtamer.rpc.loadbalancer.RandomLoadBalancer;
import me.illtamer.rpc.register.NacosServiceDiscovery;
import me.illtamer.rpc.register.ServiceDiscovery;
import me.illtamer.rpc.transport.RpcClient;
import me.illtamer.serializer.CommonSerializer;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;

@Slf4j
public class NettyClient implements RpcClient {

    private static final EventLoopGroup GROUP = new NioEventLoopGroup();
    private static final Bootstrap BOOTSTRAP = new Bootstrap()
            .group(GROUP)
            .channel(NioSocketChannel.class);

    private final ServiceDiscovery serviceDiscovery;
    private final CommonSerializer serializer;
    private final UnprocessedRequests unprocessedRequests;

    public NettyClient() {
        this(DEFAULT_SERIALIZER, new RandomLoadBalancer());
    }

    public NettyClient(int serializerCode) {
        this(serializerCode, new RandomLoadBalancer());
    }

    public NettyClient(int serializerCode, LoadBalancer loadBalancer) {
        this.serviceDiscovery = new NacosServiceDiscovery(loadBalancer);
        this.serializer = CommonSerializer.getByCode(serializerCode);
        unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class);
    }

    @Override
    public CompletableFuture<RpcResponse<?>> sendRequest(RpcRequest rpcRequest) {
        if (serializer == null) {
            log.error("未设置序列化器");
            throw new RpcException(RpcError.SERVICE_NOT_FOUND);
        }
        CompletableFuture<RpcResponse<?>> resultFuture = new CompletableFuture<>();
        try {
            // 从 Nacos 获取提供对应服务的服务端地址
            InetSocketAddress address = serviceDiscovery.lookupService(rpcRequest.getInterfaceName());
            Channel channel = ChannelProvider.get(address, serializer);
            if (!channel.isActive()) {
                GROUP.shutdownGracefully();
                return null;
            }
            // 放入新的待处理请求
            unprocessedRequests.put(rpcRequest.getRequestId(), resultFuture);
            // 向服务器发送请求并设置监听
            channel.writeAndFlush(rpcRequest).addListener((ChannelFutureListener) future1 -> {
                if(future1.isSuccess()){
                    log.info(String.format("客户端发送消息：%s", rpcRequest.toString()));
                } else {
                    future1.channel().close();
                    resultFuture.completeExceptionally(future1.cause());
                    log.error("发送消息时有错误发生:", future1.cause());
                }
            });
        } catch (Exception e) {
            unprocessedRequests.remove(rpcRequest.getRequestId());
            log.error("发送消息时有错误发生:", e);
            // 使当前线程退出阻塞状态并结束
            Thread.currentThread().interrupt();
        }
        return resultFuture;
    }

}
